package com.ywu.mq.standard.service;

import com.ywu.boot.context.YwuContext;
import com.ywu.boot.context.YwuContextHolder;
import com.ywu.boot.exception.YwuRuntimeException;
import com.ywu.common.entity.PageList;
import com.ywu.common.gson.GsonUtils;
import com.ywu.common.page.PageService;
import com.ywu.common.result.CommonResult;
import com.ywu.common.utils.TimerIdGenerateUtil;
import com.ywu.mq.standard.config.KafkaAppConfig;
import com.ywu.mq.standard.data.YwuKafkaTopicRepository;
import com.ywu.mq.standard.dto.YwuKafkaTopicDetailDto;
import com.ywu.mq.standard.dto.YwuKafkaTopicDto;
import com.ywu.mq.standard.dto.YwuKafkaTopicInnerDetailDto;
import com.ywu.mq.standard.entity.YwuKafkaTopicEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.TopicPartitionInfo;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * @ClassName YwuKafkaTopicService
 * @Description TODO
 * @Author GroundDemo
 * @Date 2025/3/27 14:31
 * @Version 1.0
 **/
@Service
@Slf4j
public class YwuKafkaTopicService {

    @Autowired
    private AdminClient adminClient;

    @Autowired
    private YwuKafkaTopicRepository repository;

    @Autowired
    private KafkaAppConfig appConfig;

    /**
    * @Author GroundDemo
    * @Description 创建主题
    * @Date 14:32 2025/3/27
    * @Param [ywuKafkaDto]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    @Transactional
    public CommonResult<Object> createKafkaTopic(YwuKafkaTopicDto ywuKafkaTopicDto) {
        // 校验主题名称是否存在
        boolean checkRes = checkTopicName(ywuKafkaTopicDto);
        if (checkRes) {
            return CommonResult.error().setMessage("topic name has exist");
        }
        try {
            // 构建entity对象
            YwuKafkaTopicEntity ywuKafkaTopicEntity = new YwuKafkaTopicEntity();
            BeanUtils.copyProperties(ywuKafkaTopicDto, ywuKafkaTopicEntity);
            YwuContext currentContext = YwuContextHolder.getCurrentContext();
            ywuKafkaTopicEntity.setId(TimerIdGenerateUtil.nextId());
            ywuKafkaTopicEntity.createInfo(currentContext.getYwuContextUser().getUserAccount());
            // 插入数据库
            repository.insertKafkaTopic(ywuKafkaTopicEntity);
            // 创建 NewTopic 对象 (主题名称需要租户隔离，所有在创建的时候带上应用ID信息)
            NewTopic newTopic = new NewTopic(getTopicName(ywuKafkaTopicEntity),
                    Integer.parseInt(ywuKafkaTopicEntity.getTopicPartitions()),
                    Short.parseShort(ywuKafkaTopicEntity.getReplicationFactor()));
            // 创建主题的请求
            CreateTopicsResult createTopicsResult = adminClient.createTopics(Collections.singleton(newTopic));
            // 等待主题创建完成
            createTopicsResult.values().get(getTopicName(ywuKafkaTopicEntity)).get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("create kafka topic error {}", e.getMessage());
            // 往上抛异常，如果出现异常，回滚事务
            throw new YwuRuntimeException("create kafka topic error");
        }
        return CommonResult.success(null);
    }


    /**
    * @Author GroundDemo
    * @Description 校验主题名称是否存在
    * @Date 10:27 2025/3/28
    * @Param [ywuKafkaTopicDto]
    * @return boolean
    **/
    private boolean checkTopicName(YwuKafkaTopicDto ywuKafkaTopicDto) {
        YwuKafkaTopicEntity search = new YwuKafkaTopicEntity();
        search.setTopicName(ywuKafkaTopicDto.getTopicName());
        search.setAppId(ywuKafkaTopicDto.getAppId());
        YwuKafkaTopicEntity queryEntity = repository.getEntity(search);
        return Objects.nonNull(queryEntity);
    }

    /**
    * @Author GroundDemo
    * @Description 查询kafka所有主题名称 - 直接调用kafka客户端查询
    * @Date 15:26 2025/3/27
    * @Param []
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    public CommonResult<Object> getKafkaSourceTopic() {
        try {
            // 发起查询主题列表的请求
            ListTopicsResult listTopicsResult = adminClient.listTopics();
            // 获取主题列表
            Set<String> topics = listTopicsResult.names().get();
            return CommonResult.success(topics);
        } catch (InterruptedException | ExecutionException e) {
            log.error("getKafkaSourceTopic query error {}", e.getMessage());
            throw new YwuRuntimeException("getKafkaSourceTopic query error");
        }
    }

    /**
    * @Author GroundDemo
    * @Description 删除主题
    * @Date 15:30 2025/3/27
    * @Param [ywuKafkaTopicDto]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    @Transactional
    public CommonResult<Object> deleteKafkaTopic(YwuKafkaTopicDto ywuKafkaTopicDto) {
        try {
            // 校验是否存在数据
            YwuKafkaTopicEntity searchEntity = new YwuKafkaTopicEntity();
            searchEntity.setId(ywuKafkaTopicDto.getId());
            YwuKafkaTopicEntity queryEntity = repository.getEntity(searchEntity);
            // 查询结果为空，返回删除失败
            if (Objects.isNull(queryEntity)) {
                return CommonResult.error().setMessage("topic is not exist");
            }
            // 删除数据表
            YwuKafkaTopicEntity deleteEntity = new YwuKafkaTopicEntity();
            deleteEntity.setId(ywuKafkaTopicDto.getId());
            repository.deleteKafkaTopicById(deleteEntity);
            // 发起删除主题的请求
            DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Collections.singleton(getTopicName(queryEntity)));
            // 等待删除操作完成
            deleteTopicsResult.all().get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("deleteKafkaTopic delete error {}", e.getMessage());
            throw new YwuRuntimeException("deleteKafkaTopic delete error");
        }
        return CommonResult.success(null);
    }

    /**
    * @Author GroundDemo
    * @Description 查询主题列表
    * @Date 15:45 2025/3/27
    * @Param [ywuKafkaTopicDto]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    public CommonResult<Object> queryKafkaTopic(YwuKafkaTopicDto ywuKafkaTopicDto) {
        PageList<YwuKafkaTopicEntity> pageListRes = new PageList<>();
        pageListRes.setPageIndex(ywuKafkaTopicDto.getPageInfo().getPageIndex());
        pageListRes.setPageSize(ywuKafkaTopicDto.getPageInfo().getPageSize());
        // 构建分页参数
        PageService.execPageStartAndEnd(ywuKafkaTopicDto.getPageInfo());
        // 如果appId没有传，返回空结果
        if (StringUtils.isEmpty(ywuKafkaTopicDto.getAppId())) {
            pageListRes.setDataList(new ArrayList<>());
            pageListRes.setRecordCount(0);
            return CommonResult.success(pageListRes);
        }
        // 查询数量
        Integer count = repository.queryListCount(ywuKafkaTopicDto);
        if (count == 0) {
            pageListRes.setDataList(new ArrayList<>());
        } else {
            List<YwuKafkaTopicEntity> ywuKafkaTopicEntities = repository.queryList(ywuKafkaTopicDto);
            // 构建一个topic和appid的复合参数，供前端使用
            ywuKafkaTopicEntities.forEach(res -> res.setTopicAppId(res.getTopicName() + "  -  " + res.getAppId()));
            if (CollectionUtils.isEmpty(ywuKafkaTopicEntities)) {
                pageListRes.setDataList(new ArrayList<>());
            } else {
                pageListRes.setDataList(ywuKafkaTopicEntities);
            }
        }
        pageListRes.setRecordCount(count);
        return CommonResult.success(pageListRes);
    }

    /**
    * @Author GroundDemo
    * @Description 由于主题需要租户隔离，所以在构造主题名称的时候把租户ID带上
    * @Date 16:02 2025/3/27
    * @Param [entity]
    * @return java.lang.String
    **/
    private String getTopicName(YwuKafkaTopicEntity entity) {
        String appId = entity.getAppId();
        String replaceAppId = appId.replace(".", "_").replace("-", "_");
        String topicName = entity.getTopicName();
        return replaceAppId + "-" + topicName;
    }

    /**
    * @Author GroundDemo
    * @Description 将主题名称还原
    * @Date 10:15 2025/3/29
    * @Param [sourceTopicName]
    * @return java.lang.String
    **/
    private String restoreTopicName(String sourceTopicName) {
        int index = sourceTopicName.indexOf("-");
        return sourceTopicName.substring(index + 1);
    }

    /**
    * @Author GroundDemo
    * @Description 修改主题信息 - 有局限性，名称不能修改，分区和副本只能添加不能修改
    * @Date 16:10 2025/3/27
    * @Param [ywuKafkaTopicDto]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    @Transactional
    public CommonResult<Object> updateKafkaTopic(YwuKafkaTopicDto ywuKafkaTopicDto) {
        // 查询原始数据
        YwuKafkaTopicEntity searchEntity = new YwuKafkaTopicEntity();
        searchEntity.setId(ywuKafkaTopicDto.getId());
        YwuKafkaTopicEntity queryEntity = repository.getEntity(searchEntity);
        // 查询结果为空，返回失败
        if (Objects.isNull(queryEntity)) {
            return CommonResult.error().setMessage("topic is not exist");
        }
        // 校验参数是否合法
        CommonResult<Object> checkRes = checkParam(ywuKafkaTopicDto, queryEntity);
        if (Objects.nonNull(checkRes)) {
            return checkRes;
        }
        // 先更新数据库
        YwuKafkaTopicEntity updateEntity = new YwuKafkaTopicEntity();
        BeanUtils.copyProperties(ywuKafkaTopicDto, updateEntity);
        YwuContext currentContext = YwuContextHolder.getCurrentContext();
        updateEntity.updateInfo(currentContext.getYwuContextUser().getUserAccount());
        repository.updateEntity(updateEntity);
        if (!ywuKafkaTopicDto.getTopicPartitions().equals(queryEntity.getTopicPartitions())) {
            // 修改分区
            doUpdateTopicPartitions(queryEntity, ywuKafkaTopicDto);
        }
        if (!ywuKafkaTopicDto.getReplicationFactor().equals(queryEntity.getReplicationFactor())) {
            // 修改副本因子
            doUpdateTopicReplicationFactor(queryEntity, ywuKafkaTopicDto);
        }
        return CommonResult.success(null);
    }

    /**
    * @Author GroundDemo
    * @Description 修改副本因子
    * @Date 18:20 2025/3/27
    * @Param [queryEntity, ywuKafkaTopicDto]
    * @return void
    **/
    private void doUpdateTopicReplicationFactor(YwuKafkaTopicEntity queryEntity, YwuKafkaTopicDto ywuKafkaTopicDto) {
        try {
            String topicName = getTopicName(queryEntity);
            // 获取主题的分区信息
            DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singleton(topicName));
            TopicDescription topicDescription = describeTopicsResult.values().get(topicName).get();
            // 生成新的分区副本分配方案
            Map<TopicPartition, Optional<NewPartitionReassignment>> newAssignments = new HashMap<>();
            // 获取Broker ID，将副本分配到broker上
            List<String> availableBrokers = Arrays.asList(appConfig.getBrokerIds().split(","));
            for (TopicPartitionInfo partitionInfo : topicDescription.partitions()) {
                TopicPartition topicPartition = new TopicPartition(topicName, partitionInfo.partition());
                List<Integer> newReplicas = new ArrayList<>();
                // 根据配置的副本因子数量生成对应下标的随机数
                List<Integer> indexList = getRandomList(Integer.parseInt(ywuKafkaTopicDto.getReplicationFactor()), availableBrokers.size());
                for (Integer index : indexList) {
                    newReplicas.add(Integer.parseInt(availableBrokers.get(index)));
                }
                NewPartitionReassignment reassignment = new NewPartitionReassignment(newReplicas);
                newAssignments.put(topicPartition, Optional.of(reassignment));
            }
            // 发起分区重新分配请求
            AlterPartitionReassignmentsResult result = adminClient.alterPartitionReassignments(newAssignments);
            // 等待重新分配完成
            result.all().get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("doUpdateTopicReplicationFactor update ReplicationFactor error {}", e.getMessage());
            throw new YwuRuntimeException("doUpdateTopicReplicationFactor update ReplicationFactor error");
        }
    }

    /**
    * @Author GroundDemo
    * @Description 生成对应数量且元素不同的随机数
    * @Date 16:56 2025/3/27
    * @Param [parseInt]
    * @return java.util.List<java.lang.Integer>
    **/
    private List<Integer> getRandomList(int num, int size) {
        List<Integer> numbers = new ArrayList<>();
        Random random = new Random();
        while (numbers.size() < num) {
            int randomNumber = random.nextInt(size);
            if (!numbers.contains(randomNumber)) {
                numbers.add(randomNumber);
            }
        }
        return numbers;
    }

    /**
    * @Author GroundDemo
    * @Description 修改分区
    * @Date 16:24 2025/3/27
    * @Param [queryEntity, ywuKafkaTopicDto]
    * @return void
    **/
    private void doUpdateTopicPartitions(YwuKafkaTopicEntity queryEntity, YwuKafkaTopicDto ywuKafkaTopicDto) {
        try {
            Map<String, NewPartitions> newPartitionsMap = new HashMap<>();
            newPartitionsMap.put(getTopicName(queryEntity), NewPartitions.increaseTo(Integer.parseInt(ywuKafkaTopicDto.getTopicPartitions())));
            CreatePartitionsResult result = adminClient.createPartitions(newPartitionsMap);
            result.all().get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("doUpdateTopicPartitions update Partitions error {}", e.getMessage());
            throw new YwuRuntimeException("doUpdateTopicPartitions update Partitions error");
        }
    }

    /**
    * @Author GroundDemo
    * @Description 校验参数 - 分区和副本只能添加不能修改
    * @Date 16:14 2025/3/27
    * @Param [dto, entity]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    private CommonResult<Object> checkParam(YwuKafkaTopicDto dto, YwuKafkaTopicEntity entity) {
        if (!StringUtils.isEmpty(dto.getTopicPartitions())) {
            // 校验分区，如果传入的参数与原始值一样，不需要处理，如果不一样，判断是否小于原来的值，小于的话则异常
            if (!dto.getTopicPartitions().equals(entity.getTopicPartitions()) &&
                    Integer.parseInt(dto.getTopicPartitions()) < Integer.parseInt(entity.getTopicPartitions())) {
                return CommonResult.error().setErrorCode("topicPartitions.can.not.less.than.source.value")
                        .setMessage("主题分区不能小于原始数值");
            }
        }
        if (!StringUtils.isEmpty(dto.getReplicationFactor())) {
            // 校验副本因子
            if (!dto.getReplicationFactor().equals(entity.getReplicationFactor()) &&
                    Integer.parseInt(dto.getReplicationFactor()) < Integer.parseInt(entity.getReplicationFactor())) {
                return CommonResult.error().setErrorCode("replicationFactor.can.not.less.than.source.value")
                        .setMessage("主题副本因子不能小于原始数值");
            }
        }
        return null;
    }

    /**
    * @Author GroundDemo
    * @Description 查询主题详情
    * @Date 8:31 2025/3/28
    * @Param [ywuKafkaTopicDto]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    public CommonResult<Object> queryKafkaTopicDetail(YwuKafkaTopicDto ywuKafkaTopicDto) {
        // 查询主题是否存在
        YwuKafkaTopicEntity searchEntity = new YwuKafkaTopicEntity();
        searchEntity.setId(ywuKafkaTopicDto.getId());
        YwuKafkaTopicEntity queryEntity = repository.getEntity(searchEntity);
        // 查询结果为空，返回删除失败
        if (Objects.isNull(queryEntity)) {
            return CommonResult.error().setMessage("topic is not exist");
        }
        try {
            String topicName = getTopicName(queryEntity);
            // 获取主题的分区信息
            DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singleton(topicName));
            TopicDescription topicDescription = describeTopicsResult.values().get(topicName).get();
            // 构建返回结果
            return buildReturnDetail(topicDescription, queryEntity);
        } catch (InterruptedException | ExecutionException e) {
            log.error("queryKafkaTopicDetail query details error {}", e.getMessage());
            throw new YwuRuntimeException("queryKafkaTopicDetail query details error");
        }
    }

    /**
    * @Author GroundDemo
    * @Description 构建主题详情返回信息
    * @Date 8:53 2025/3/28
    * @Param [topicDescription, queryEntity]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    private CommonResult<Object> buildReturnDetail(TopicDescription topicDescription, YwuKafkaTopicEntity queryEntity) {
        YwuKafkaTopicDetailDto res = new YwuKafkaTopicDetailDto();
        res.setPartitionCount(queryEntity.getTopicPartitions());
        res.setReplicationFactor(queryEntity.getReplicationFactor());
        res.setTopicId(topicDescription.topicId().toString());
        res.setTopicName(restoreTopicName(topicDescription.name()));
        List<YwuKafkaTopicInnerDetailDto> detailDtoList = new ArrayList<>();
        // 获取结点id与结点名称的映射
        String brokerMap = appConfig.getBrokerMap();
        Map brokerMapInfo = GsonUtils.deserialize(brokerMap, Map.class);
        res.setDetailDtoList(detailDtoList);
        for (TopicPartitionInfo partitionInfo : topicDescription.partitions()) {
            YwuKafkaTopicInnerDetailDto tempDto = new YwuKafkaTopicInnerDetailDto();
            tempDto.setTopicName(restoreTopicName(topicDescription.name()));
            tempDto.setPartition(String.valueOf(partitionInfo.partition()));
            // 设置分区主结点信息
            Node leader = partitionInfo.leader();
            if (brokerMapInfo.containsKey(leader.idString())) {
                tempDto.setLeaderNode(brokerMapInfo.get(leader.idString()).toString());
            } else {
                tempDto.setLeaderNode(leader.idString());
            }
            // 设置分区副本结点信息
            List<Node> replicas = partitionInfo.replicas();
            StringBuilder tempBuilder = new StringBuilder();
            for (Node node : replicas) {
                if (brokerMapInfo.containsKey(node.idString())) {
                    tempBuilder.append(",").append(brokerMapInfo.get(node.idString()).toString());
                } else {
                    tempBuilder.append(",").append(node.idString());
                }
            }
            // 去除最前面的逗号
            if (tempBuilder.length() > 0) {
                tempDto.setReplicasNode(tempBuilder.substring(1));
            } else {
                tempDto.setReplicasNode(tempBuilder.toString());
            }
            detailDtoList.add(tempDto);
        }
        // 适配前端表格展示，返回列表
        ArrayList<YwuKafkaTopicDetailDto> returnList = new ArrayList<>();
        returnList.add(res);
        return CommonResult.success(returnList);
    }
}
